package com.shujia.transformation

import org.apache.flink.streaming.api.scala._

object Demo06Aggregations {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)

    // 读取students.txt
    val stuDS: DataStream[String] = env.readTextFile("Flink/data/stu/students.txt")

    // 每个班级年龄最大值
    stuDS
      .map(line => {
        val stuArr: Array[String] = line.split(",")
        (stuArr(0), stuArr(1), stuArr(2), stuArr(3), stuArr(4))
      })
      .keyBy(_._5)
      .max(2)
//      .print()

    /**
     * max min 只能找出年龄的最大值 其他的数据以第一条数据为准
     * maxBy minBy 既可以找出年龄的最大值 其他的数据以截至到当前行的数据 第一次出现 最大或最小的数据为准
     */

    // 每个班级年龄最大值
    stuDS
      .map(line => {
        val stuArr: Array[String] = line.split(",")
        (stuArr(0), stuArr(1), stuArr(2), stuArr(3), stuArr(4))
      })
      .keyBy(_._5)
      .maxBy(2)
      .print()

    env.execute()

  }

}
